package com.enlorenz.core.channel.activemq;

import java.io.IOException;
import java.io.Serializable;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

import com.enlorenz.core.channel.IChannelEngine;
import com.enlorenz.core.engine.exception.EngineControlException;
import com.enlorenz.core.queue.proc.AbstractObjectCycleThread;

/**
 * MQ客户端，用于发送消息
 * @author MAC
 */
public class MQClientChannel extends AbstractObjectCycleThread implements IChannelEngine{
	private static final String maxInactivityDuration = "?wireFormat.maxInactivityDuration=0";
	//登陆用户名
	private String userName;
	//登陆密码
	private String userPwd;
	//MQ服务地址
	private String mqUrl;
	//创建通道类型
	private String mqName;
	//是否持久化
	private boolean isPersist=false;
	//是否开启事物
	private boolean isTransacted = false; 
	//链接
	private Connection connection; 
	//通道
	private Destination destination; 
	//生产者
	private MessageProducer producer; 
	//会话
	private Session session; 
	//发送是否成功
	private boolean sendFlag=true;
	

	/**
	 * 初始化控制线程
	 */
	protected void initEngine() throws EngineControlException{
		super.initEngine();
	}
	
	/**
	 * 启动MQ客户端
	 */
	protected void startEngine() throws EngineControlException {
		try {
			//启动服务
			clientStart();
			super.startEngine();
		} catch (Exception e) {
			e.printStackTrace();
			close();
			new EngineControlException(this.getName(),e);
		}
	}
	
	/**
	 * MQ客户端启动
	 * 
	 * @throws IOException
	 */
	private void clientStart() throws Exception {
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				userName, userPwd, mqUrl + maxInactivityDuration);
		// 构造从工厂得到连接对象
		connection = connectionFactory.createConnection();
		// 启动
		connection.start();
		// 获取操作连接
		session = connection.createSession(isTransacted,
				Session.AUTO_ACKNOWLEDGE);
		destination = session.createQueue(mqName);
		producer = session.createProducer(destination);
		if (isPersist) {
			producer.setDeliveryMode(DeliveryMode.PERSISTENT);
		} else {
			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		}
	}
	
	/**
	 * 关闭打开的链接
	 */
	private void close() {
		if(null!=producer){
			try {
				producer.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
		if(null!=session){
			try {
				session.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
		if(null!=connection){
			try {
				connection.stop();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
		if(null!=connection){
			try {
				connection.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
	}
	
	/**
	 * 关闭Socket服务
	 */
	@Override
	protected void shutdownEngine() throws EngineControlException {
		//关闭选择器和服务
		close();
		super.shutdownEngine();
	}
	
	/**
	 * Socket是否还存活
	 */
	@Override
	public boolean isAlive() {
		return sendFlag && super.isAlive();
	}
	
	
	@Override
	protected void process(Object result) {
		try {
			ObjectMessage msg = session.createObjectMessage();
			msg.setObject((Serializable)result);
			producer.send(msg);
			if (isTransacted)
			{
				session.commit();
			}
			sendFlag=true;
		} catch (JMSException e) {
			e.printStackTrace();
			sendFlag=false;
		}
	} 

	public String getMqName() {
		return mqName;
	}

	public void setMqName(String mqName) {
		this.mqName = mqName;
	}

	public void setMqUrl(String mqUrl) {
		this.mqUrl = mqUrl;
	}
}
